package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TableProcess;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.utils.HbaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

import java.util.ArrayList;

public class DimHbaseSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection hbaseConnection;
    @Override
    public void open(Configuration parameters) throws Exception {
         hbaseConnection = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConnection);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        //读取数据的操作类型
        String type = jsonObj.getString("type");

        TableProcess tableProcess = jsonObj.getObject("TableProcess", TableProcess.class);
        String sinkTable = tableProcess.getSinkTable();
        String sinkFamily = tableProcess.getSinkFamily();
        String sinkRowKey = tableProcess.getSinkRowKey();
        String row = jsonObj.getString(sinkRowKey);
        //判断操作类型是写入还是删除
        if("delete".equals(type)){

            HbaseUtil.delRow(hbaseConnection, EduConfig.HBASE_NAMESPACE,sinkTable,row);
        }else {
            String[] columns = tableProcess.getSinkColumns().split(",");
            String[] values = new String[columns.length];
            for (int i = 0; i < columns.length; i++) {
                String value = jsonObj.getString(columns[i]);
                values[i]=value;
            }
            HbaseUtil.putRow(hbaseConnection, EduConfig.HBASE_NAMESPACE,sinkTable,row,sinkFamily,columns,values);
        }

    }
}
